package com.study.rocket.transaction.producer;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * @author mjy(针对消息事务的处理)
 * @date 2021/12/14
 * @description
 */
public class Producer {

    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("transactionGroup");
        producer.setNamesrvAddr("192.168.15.170:9876");
        //设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 执行本地事务
             * @param message
             * @param o
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if (message.getTags().equals("tag1")) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (message.getTags().equals("tag2")) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }else {
                    return LocalTransactionState.UNKNOW;
                }
            }

            /**
             * 该方法是mq进行消息事务状态的一个回查
             * @param messageExt
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();
        System.out.println("消息者启动成功。。。。。。。。。");
        for (int i = 1; i <= 3; i++) {
            Message message = new Message("transactionTopic", "tag" + i, ("事务消息" + i).getBytes());
            TransactionSendResult rest = producer.sendMessageInTransaction(message, null);
            System.out.println("消息发送"+rest);
        }
    }
}
